package com.wasp.collector.service.akka.cluster;

import java.util.HashSet;
import java.util.Set;

import com.wasp.collector.service.akka.cluster.IClusterClient;

import akka.actor.AbstractActor;
import akka.actor.Address;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent.MemberEvent;
import akka.cluster.ClusterEvent.MemberUp;
import akka.cluster.ClusterEvent.ReachabilityEvent;
import akka.cluster.ClusterEvent.ReachableMember;
import akka.cluster.ClusterEvent.UnreachableMember;
import akka.japi.pf.ReceiveBuilder;
import scala.PartialFunction;
import scala.runtime.BoxedUnit;

public class ClusterClient extends AbstractActor{

	final Set<Address> nodes = new HashSet<Address>();
	Cluster cluster = Cluster.get(getContext().system());

	public ClusterClient() {

	}
	@Override
	public void preStart() {
		cluster.subscribe(self(), MemberEvent.class, ReachabilityEvent.class);
	}

	// re-subscribe when restart
	@Override
	public void postStop() {
		cluster.unsubscribe(self());
	}

	@Override
	public PartialFunction<Object, BoxedUnit> receive() {
		PartialFunction<Object, BoxedUnit> function = ReceiveBuilder.match(MemberUp.class, mUp -> {
			if (mUp.member().hasRole(AkkaService.ROLE))
				nodes.add(mUp.member().address());
		}).match(MemberEvent.class, other -> {
			nodes.remove(other.member().address());
		}).match(UnreachableMember.class, unreachable -> {
			nodes.remove(unreachable.member().address());
		}).match(ReachableMember.class, reachable -> {
			if (reachable.member().hasRole(AkkaService.ROLE))
				nodes.add(reachable.member().address());
		}).build();
		refreshNodes();
		return function;
	}
	public void refreshNodes() {
		ClusterClientHelper.setNodes(nodes);
	}

}
